package asteroid

import (
	"sync"

	"gddgame.cc/galaxy/utils"
	"gddgame.cc/galaxy/utils/def"
	"github.com/antlabs/timer"

	"github.com/pkg/errors"
)

var (
	LoggerInvalidErr = errors.New("Logger is invalid")
)

type LayerInterface interface {
	// 用于建立同层通信
	Region(region string) (*Region, bool)

	Dispatch(msg []byte)
	Track(key string)
	CloseChan() chan struct{}
}

type Layer struct {
	def.Logger

	node     string
	trackKey string

	regions   sync.Map
	regionMux sync.Mutex
	Channel   channel
	Proxy

	socket Socket

	regionAdapt Adapt

	close     chan struct{}
	needClose bool

	pipePool sync.Pool
	*utils.ThreaderPool
	timer timer.Timer

	// 等待退出的节点
	exitNodes []string
	exit      bool
}

func NewLayer(node string, logger def.Logger, timer timer.Timer) (*Layer, error) {
	l := &Layer{
		Logger:    logger,
		node:      node,
		regions:   sync.Map{},
		regionMux: sync.Mutex{},

		socket: nil,

		close:     make(chan struct{}),
		needClose: true,

		ThreaderPool: utils.NewThreaderPool(nil, 30, 10),
		timer:        timer,
		pipePool:     sync.Pool{},

		trackKey: "trackId",

		exitNodes: []string{},
		exit:      false,
	}
	l.Channel = channel{
		layer:  l,
		Logger: l.LoggerWrap("Service", "Channel"),
		node:   l.node,
		adapt:  &LocalAdapt{},
		mu:     sync.Mutex{},
	}
	l.Proxy = Proxy{
		layer: l,
	}
	l.regionAdapt = &LocalAdapt{}
	l.socket = &LocalSocket{}
	l.pipePool.New = l.newPipe
	return l, nil
}

func (l *Layer) newPipe() interface{} {
	return newPipe(l)
}

func (l *Layer) Pipe() *Pipe {
	return l.pipePool.Get().(*Pipe)
}

func (l *Layer) Track(key string) {
	l.trackKey = key
}

func (l *Layer) Region(key string) (*Region, bool) {
	l.regionMux.Lock()
	var region *Region
	if i, ok := l.regions.Load(key); ok {
		l.regionMux.Unlock()
		return i.(*Region), true
	}
	region = newRegion(l, l.node, key)
	l.regions.Store(key, region)
	l.regionMux.Unlock()
	return region, false
}

func (l *Layer) dispatch(msgB []byte) {
	//fmt.Println("dispatch", msgB)
	msg, err := newAccept(msgB, l)
	if err != nil {
		return
	}
	//if msg.Key == "core" {
	//	fmt.Println("core recv", l.node)
	//}
	switch msg.MessageType {
	case RegionMessage:
		//l.Debugf("accept Node message:%s", msg.From)
		i, ok := l.regions.Load(msg.Logic)
		if ok {
			i.(*Region).dispatch(msg)
		}
	case ChannelMessage:
		l.Channel.dispatch(msg)
	case ProxyMessage:
		l.Proxy.dispatch(msg)
	case EnterMessage:
		l.onEnter(msg.GetFrom())
	case ExitMessage:
		l.OnExit(msg.GetFrom())
	}
	msg.Release()
}

func (l *Layer) ConnectNode(node string) error {
	if err := l.socket.Connect(node); err != nil {
		return err
	}
	// 发送给新链接节点，enter事件
	msg := newMsg(0, "", "")
	msg.to = []string{node}
	msg.From = l.node
	msg.layer = l
	return msg.Send(EnterMessage, nil)
}

func (l *Layer) SetSocket(socket Socket) error {
	socket.Bind(l.dispatch, l.close)
	if err := socket.Listen(l.node); err != nil {
		return err
	}
	l.socket = socket
	return nil
}

func (l *Layer) SetRegionAdapt(adapt Adapt) error {
	l.regionAdapt = adapt
	return nil
}

func (l *Layer) StartChannel(adapt Adapt) error {
	return l.Channel.setAdapt(adapt)
}

func (l *Layer) CloseChan() chan struct{} {
	return l.close
}

// 安全退出操作是通过不分配给该节点信息实现的
// 节点任务分配是通过Strategy进行处理
// 所以通过nodeManager在给Strategy传递节点信息是过滤退出中node
func (l *Layer) Exit() error {
	l.exit = true
	l.OnExit(l.node)
	nodes := l.socket.Nodes()
	// 通知其他节点，本节点准备退出
	msg := newMsg(0, "", "")
	msg.to = nodes
	msg.From = l.node
	msg.layer = l
	return msg.Send(ExitMessage, nil)
}

func (l *Layer) OnExit(node string) {
	nodes, ok := utils.AddString(l.exitNodes, node)
	if ok {
		l.exitNodes = nodes
		l.resetNodeManager()
	}
}

func (l *Layer) onEnter(node string) {
	nodes, ok := utils.RemoveString(l.exitNodes, node)
	if ok {
		l.exitNodes = nodes
		l.resetNodeManager()
	}
}

func (l *Layer) resetNodeManager() {
	l.regions.Range(func(key, value interface{}) bool {
		region := value.(*Region)
		region.setStrategyNodes()
		region.rooms.Range(func(key, value interface{}) bool {
			room := value.(*Room)
			room.setStrategyNodes()
			return true
		})
		return true
	})
	l.Channel.publishers.Range(func(key, value interface{}) bool {
		publisher := value.(*Publisher)
		publisher.setStrategyNodes()
		return true
	})
}

func (l *Layer) Close() (err error) {
	if !l.needClose {
		return nil
	}
	l.needClose = false
	close(l.close)
	l.ThreaderPool.Clean()
	//l.Debugf("[ Asteroid ] Close Region")
	l.regions.Range(func(key, value interface{}) bool {
		err = value.(*Region).Close()
		return true
	})
	//l.Debugf("[ Asteroid ] Close Channel")
	if err := l.Channel.close(); err != nil {
		return err
	}

	//l.Debugf("[ Asteroid ] Close Socket")
	if l.socket != nil {
		if err := l.socket.Close(); err != nil {
			return err
		}
		l.socket = nil
	}
	//l.Debugf("[ Asteroid ] Clean Thread")
	l.Clean()
	return nil
}
